package riwcMajority;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import atomicRegister.*;

import BEB.*;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pDeliver;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;

/**
 * The <code>SimpleUnreliableBroadcast</code> class.
 * 
 * @author Gianmario Spacagna <spacagna@kth.se>
 */
public final class RIWCMajority extends ComponentDefinition {

	Positive<BEB> beb = requires(BEB.class);
	Positive<PerfectPointToPointLink> pp2p = requires(PerfectPointToPointLink.class);
	Negative<AtomicRegister> ar = provides(AtomicRegister.class);

	private static final Logger logger = LoggerFactory
			.getLogger(RIWCMajority.class);

	private Set<Address> neighborSet;
	private Address self;

	private int maxR;

	private ArrayList<HashSet<Address>> writeSet;
	private ArrayList<HashSet<SetItem>> readSet;
	private boolean[] reading;
	private int[] reqid;
	private int[] v;
	private long[] ts;
	private int[] mrank;
	private int[] writeval;
	private int[] readval;

	/**
	 * Instantiates a new application0.
	 */
	public RIWCMajority() {
		subscribe(handleInit, control);
		subscribe(handleStart, control);

		subscribe(handleNNRegRead, ar);
		subscribe(handleNNRegWrite, ar);
		subscribe(handleReadMessage, beb);
		subscribe(handleWriteMessage, beb);
		subscribe(handleReadValMessage, pp2p);
		subscribe(handleAckMessage, pp2p);

	}

	Handler<AckMessage> handleAckMessage = new Handler<AckMessage>() {

		@Override
		public void handle(AckMessage event) {
			int r = event.getR();
			if (event.getId() == reqid[r]) {
				writeSet.get(r).add(event.getSource());
			}
			writeMajorityReached(r);
		}
	};
	Handler<ReadMessage> handleReadMessage = new Handler<ReadMessage>() {

		@Override
		public void handle(ReadMessage event) {
			int r = event.getR();
			trigger(new ReadValMessage(self, r, event.getReqid(), new SetItem(
					ts[r], mrank[r], v[r])), pp2p);

		}
	};

	Handler<WriteMessage> handleWriteMessage = new Handler<WriteMessage>() {

		@Override
		public void handle(WriteMessage event) {
			int r = event.getR();
			if (event.getSi().getT() > ts[r]
					|| (event.getSi().getT() == ts[r] && event.getSi().getRk() > mrank[r])) {
				v[r] = event.getSi().getVal();
				ts[r] = event.getSi().getT();
				mrank[r] = event.getSi().getRk();
			}
			trigger(new AckMessage(self, r, event.getReqid()), pp2p);
		}
	};

	Handler<ReadValMessage> handleReadValMessage = new Handler<ReadValMessage>() {

		@Override
		public void handle(ReadValMessage event) {
			int r = event.getR();
			if (event.getId() == reqid[r]) {
				readSet.get(r).add(event.getSi());
			}
			readMajorityReached(r);
		}

	};

	Handler<WriteRequest> handleNNRegWrite = new Handler<WriteRequest>() {

		@Override
		public void handle(WriteRequest event) {
			int r = event.getR();
			reqid[r]++;
			writeval[r] = event.getVal();
			readSet.clear();
			writeSet.clear();
			trigger(new ReadMessage(self, r, reqid[r]), beb);

		}

	};

	Handler<ReadRequest> handleNNRegRead = new Handler<ReadRequest>() {

		@Override
		public void handle(ReadRequest event) {
			// do something when readrequest happen
			int r = event.getR();
			reqid[r]++;
			reading[r] = true;
			readSet.clear();
			writeSet.clear();

			trigger(new ReadMessage(self, r, reqid[r]), beb);

		}

	};

	Handler<RIWCMajorityInit> handleInit = new Handler<RIWCMajorityInit>() {
		public void handle(RIWCMajorityInit event) {
			logger.info("Received Init event");
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			maxR = event.getMaxR();

			writeSet = new ArrayList<HashSet<Address>>();
			readSet = new ArrayList<HashSet<SetItem>>();
			reading = new boolean[maxR];
			reqid = new int[maxR];
			v = new int[maxR];
			ts = new long[maxR];
			mrank = new int[maxR];
			writeval = new int[maxR];
			readval = new int[maxR];

			for (int r = 0; r < maxR; r++) {
				writeSet.add(new HashSet<Address>());
				readSet.add(new HashSet<SetItem>());
				reading[r] = false;
				reqid[r] = 0;
				v[r] = 0;
				ts[r] = 0;
				mrank[r] = 0;
			}

		}
	};

	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
		}
	};

	void readMajorityReached(int r) {
		// check if majority is reached
		if (writeSet.get(r).size() > neighborSet.size() / 2) {

			SetItem highest = null;
			// take the first item from the set
			for (SetItem si : readSet.get(r)) {
				highest = si;
				break;

			}
			for (SetItem si : readSet.get(r)) {
				if (si.getT() > highest.getT())
					highest = si;
				else if (si.getT() == highest.getT()) {
					if (si.getRk() > highest.getRk())
						highest = si;
				}
			}
			readval[r] = highest.getVal();

			if (reading[r] == true) {
				trigger(new WriteMessage(self, r, reqid[r], highest), beb);
			} else {
				trigger(new WriteMessage(self, r, reqid[r], new SetItem(highest
						.getT() + 1, mrank[r], writeval[r])), beb);
			}
		}

	}

	void writeMajorityReached(int r) {
		if (writeSet.get(r).size() > neighborSet.size() / 2) {
			if (reading[r] == true) {
				reading[r] = false;
				trigger(new ReadResponse(r,readval[r]), ar);
			} else {
				trigger(new WriteResponse(r), ar);
			}
		}
	}

	@SuppressWarnings("unused")
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
